package com.dmp.report

import com.dmp.beans.Logs
import com.dmp.utils.ReportUtils
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 3.2.3 媒体分析(App)报告
  * 传入参数：
  * G:\光环国际大数据开发班\大数据最后阶段-项目\21-dmp项目\资料\data.txt G:\光环国际大数据开发班\大数据最后阶段-项目\21-dmp项目\资料\appmapping.txt xx
  *
  * 运行结果：
  * 马上赚 0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0
  * 其他 2.0,2.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0
  */
object AppReport {
  def main(args: Array[String]): Unit = {
    /**
      * 1) 判断参数个数
      */
    if (args.length < 3) {
      println(
        """
          |com.dmp.report.AppReport <logDataPath> <appMappingPath> <outputPath>
          | <logDataPath> 日志目录
          | <appMappingPath> 映射文件目录
          | <outputPath> 输出结果文件目录
        """.stripMargin)
      System.exit(0)
    }
    /**
      * 2）接收参数
      */
    val Array(logDataPath, appMappingPath, outpoutPath) = args
    /**
      * 3) 初始化程序入口
      */
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.setMaster("local")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.registerKryoClasses(Array(classOf[Logs]))
    val sc = new SparkContext(conf)
    /**
      * 4) 把APP映射文件作为广播变量
      * 数据格式：
      * 1	乐自游	A06		cn.net.inch.android	通过GPS的定为实现景区的自动语音讲解的功能。
      * 可能出现的原因：APP名字可能会产生变化，但是APPID号不会变化是唯一的，所以需要映射
      */
    val appMap: Map[String, String] = sc.textFile(appMappingPath).flatMap(line => {
      import scala.collection.mutable.Map
      val map = Map[String, String]()
      val fields: Array[String] = line.split("\t")
      map += (fields(4) -> fields(1))
      map
    }).collect().toMap

    val broadcastAppMap = sc.broadcast(appMap)

    /**
      * 5) 生成报表
      */
    sc.textFile(logDataPath).map(line => {
      val log = Logs.line2Log(line)
      val adRequest = ReportUtils.calculateRequest(log)
      val adResponse = ReportUtils.calculateResponse(log)
      val adClick = ReportUtils.calculateShowClick(log)
      val adCost = ReportUtils.calculateAdCost(log)
      //统计的媒体APP，.value就获取到值了。Map里面有getOrElse功能，拿着log.appid去获取映射里面的值，如果能获取到就用这个值，如果获取不到就使用log.appname
      val appName = broadcastAppMap.value.getOrElse(log.appid, log.appname)
      //List(1,1) ++ List(0,0)  => List(1,1,0,0)
      (appName, adRequest ++ adResponse ++ adClick ++ adCost)
    }).filter(tuple => {
      tuple._1.nonEmpty && !"".equals(tuple._1)
    }).reduceByKey {
      case (list1, list2) => {
        //List(1,0) .zip List(2,3)  => List((1,2),(0,3))
        list1.zip(list2).map {
          case (x, y) => x + y
        }
      }
    }.foreach(tuple => {
      val appName = tuple._1
      val report = tuple._2.mkString(",")
      println(appName + " " + report)
    })


    sc.stop()

  }

}
